package cn.pengpeng.udaf;

import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;

/**
 * 1 init()方法  init（）方法负责初始化计算函数并重设它的内部状态。
 * 在MaximumIntUDAFEvaluator中，我们把存放最终结果的IntWritable对象设置为null。
 * 我们使用null来表示目前还没有对任何值进行聚集计算，这和对空集NULL计算最大值应有的结果是一致的。
 * 
 * 2 iterate（）方法 每次对一个新值进行聚集计算时都会调用iterate（）方法。
 * 计算函数要根据聚集计算的结果更新其内部状态。iterate（）接受的参数和Hive中被调用函数的参数使对应的。
 * 
 * 3 terminatePartial（）方法  Hive需要部分聚集结果时会调用terminatePartial（）方法。
 * 这个方法必须返回一个封装了聚集计算当前状态的对象。
 * 
 * 4 merge（）方法  在Hive决定要合并一个部分聚集值和另一个部分聚集值时会调用merge（）方法。
 * 该方法接受一个对象作为输入。这个对象的类型必须和terminatePartial（）方法返回的类型一致。
 * 
 * 5 terminate（） Hive需要最终聚集结果时会调用terminate（）方法。
 * 计算函数需要把状态作为一个值返回。
 * 
 * 
 * 
 * @author root
 *
 */


public class MyUDAF implements UDAFEvaluator{
	public static class PartialResult{
		String skuids;
		String delimiter;
	}
	
	private PartialResult partial;
	
	@Override
	public void init() {
		partial = null;
	}
	
	public boolean iterate(String item_sku_id,String deli){
		if(item_sku_id == null){
			return true;
		}
		
		if(partial == null){
			partial = new PartialResult();
			partial.skuids = new String("");
			if(deli == null || deli.equals("")){
				partial.delimiter = new String(",");
			}else{
				partial.delimiter = new String(deli);
			}
		}
		if (partial.skuids.length() >0){
			partial.skuids = partial.skuids.concat(partial.delimiter);
		}
		
		partial.skuids = partial.skuids.concat(item_sku_id);
		return true;
	}
	
	public PartialResult terminatePartial(){
		return partial;
	}
	
	public boolean merge(PartialResult other){
		if(other == null){
			return true;
		}
		if(partial == null){
			partial = new PartialResult();
			partial.skuids = new String(other.skuids);
			partial.delimiter = new String(other.delimiter);
		}else{
			 if (partial.skuids.length() > 0) {
                 partial.skuids = partial.skuids.concat(partial.delimiter);
             }
             partial.skuids = partial.skuids.concat(other.skuids);
		}
		return true;
	}
	
	 public String terminate() {
         return new String(partial.skuids);
     }

}
